package cn.ac.iie.di.ban.ycl.cleaner;

import cn.ac.iie.di.ban.ycl.cleaner.common.SendObject;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.ban.ycl.cleaner.handler.CleanerHandler;
import cn.ac.iie.di.ban.ycl.cleaner.handler.KillHandler;
import cn.ac.iie.di.ban.ycl.cleaner.receiver.MessageFailReceiver;
import cn.ac.iie.di.ban.ycl.cleaner.receiver.MessageSuccessReceiver;
import cn.ac.iie.di.ban.ycl.cleaner.sender.DeduplicationMessageSenderThread;
import cn.ac.iie.di.ban.ycl.cleaner.sender.FilterMessageSenderThread;
import cn.ac.iie.di.ban.ycl.cleaner.sender.RubbishMessageSenderThread;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import java.io.File;
import java.util.concurrent.LinkedBlockingDeque;
import org.apache.commons.configuration2.FileBasedConfiguration;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.builder.FileBasedConfigurationBuilder;
import org.apache.commons.configuration2.builder.fluent.Parameters;
import org.apache.commons.configuration2.convert.DefaultListDelimiterHandler;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.log4j.PropertyConfigurator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @date 2016-9-8 14:32:24
 * @version V1.0
 * @desc 程序入口
 */
public class CleanerMain {

    private static final Logger LOG = LoggerFactory.getLogger(CleanerHandler.class);
    public static LinkedBlockingDeque<REMessage> handlerQueue;//接收队列
    public static LinkedBlockingDeque<REMessage> deduplicationQueue;//去重队列
    public static LinkedBlockingDeque<REMessage> filterQueue;//过滤队列
    public static LinkedBlockingDeque<REMessage> rubbishQueue;//垃圾队列
    static final String PROPERTIES_PATH;

    static {
        String projectHome = System.getenv("PROJECT_HOME");
        if (projectHome == null || projectHome.isEmpty()) {
            throw new RuntimeException("please set project home!");
        }
        String log4jPath = projectHome + File.separator + "conf/log4j.properties";
        if (log4jPath == null || log4jPath.isEmpty()) {
            throw new RuntimeException("log4j path is empty!");
        } else {
            LOG.info("LOG4J_PATH = {}", log4jPath);
            PropertyConfigurator.configure(log4jPath);
        }
        PROPERTIES_PATH = projectHome + File.separator + "conf/configuration.properties";
        if (PROPERTIES_PATH == null || PROPERTIES_PATH.isEmpty()) {
            throw new RuntimeException("properties path is empty!");
        } else {
            LOG.info("PROPERTIES_PATH = {}", PROPERTIES_PATH);
        }
    }

    public static void main(String[] args) throws REConnectionException, Exception {
        //读取配置文件
        FileBasedConfiguration configuration = initConfiguration();

        //启动数据发送
        deduplicationQueue = new LinkedBlockingDeque<>(10);
        filterQueue = new LinkedBlockingDeque<>(10);
        rubbishQueue = new LinkedBlockingDeque<>(10);
        boolean isNeedDeduplicate = configuration.getBoolean(Constants.DEDUPLICATION_NECESSARY);
        if (isNeedDeduplicate) {
            Thread deduplicationSendThread = new Thread(
                    new DeduplicationMessageSenderThread(configuration), "deduplication send thread");
            deduplicationSendThread.start();
        }
        boolean isNeedFilter = configuration.getBoolean(Constants.FILTER_NECESSARY);
        if (isNeedFilter) {
            Thread filterSendThead = new Thread(
                    new FilterMessageSenderThread(configuration), "filter send thread");
            filterSendThead.start();
            Thread garbageSendThread = new Thread(
                    new RubbishMessageSenderThread(configuration), "rubbish send thread");
            garbageSendThread.start();
        }

        //启动数据清洗
        int handlerQueueSize = configuration.getInt(Constants.HANDLER_QUEUE_SIZE, 10);
        int handlerThreadNum = configuration.getInt(Constants.HANDLER_THREAD_NUM, 4);
        int handlerBathNum = configuration.getInt(Constants.HANDLER_BATCH_NUM, 10);
        int handlerTimeOut = configuration.getInt(Constants.HANDLER_TIME_OUT);

        LOG.info("--------------------------------------------");
        LOG.info("   handler queue size = " + handlerQueueSize);
        LOG.info("   handler thread num = " + handlerThreadNum);
        LOG.info("   handler  batch num = " + handlerBathNum);
        LOG.info("     handler time out = " + handlerTimeOut);
        LOG.info("--------------------------------------------");

        handlerQueue = new LinkedBlockingDeque<>(handlerQueueSize);
        for (int i = 0; i < handlerThreadNum; i++) {
            Thread thread = new Thread(new CleanerHandler(configuration, handlerQueue));
            thread.start();
        }

        //启动数据接收
        String mqNameserver = configuration.getString(Constants.RECEIVER_MQ_NAMESERVER);
        String mqTopic = configuration.getString(Constants.RECEIVER_MQ_READ_TOPIC);
        String mqGroup = configuration.getString(Constants.RECEIVER_MQ_GROUP);
        int consumerThreadNum = configuration.getInt(Constants.RECEIVER_THREAD_NUM, 4);
        LOG.info("--------------------------------------------");
        LOG.info("   receiver mq nameserver = " + mqNameserver);
        LOG.info("        receiver mq topic = " + mqTopic);
        LOG.info("       receiver  mq group = " + mqGroup);
        LOG.info("      receiver thread num = " + consumerThreadNum);
        LOG.info("--------------------------------------------");
        REConnection conn = new REConnection(mqNameserver);
        REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn
                .getReceiveSessionBuilder(mqTopic);
        builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
        builder.setConsumeThreadNum(consumerThreadNum);
        builder.setHandler(new MessageSuccessReceiver(handlerQueue));
        builder.setFailureHandler(new MessageFailReceiver());
        builder.setGroupName(mqGroup);
        REReceiveSession session = (REReceiveSession) builder.build();
        session.start();

        //程序关闭
        KillHandler killHandler = new KillHandler();
        killHandler.registerSignal("TERM");
        killHandler.putQueue("deduplication queue", deduplicationQueue);
        killHandler.putQueue("filter queue", deduplicationQueue);
        killHandler.putQueue("rubbish queue", rubbishQueue);
        killHandler.setConn(conn);
        LOG.info("ban ycl cleaner start success!");
    }

    private static FileBasedConfiguration initConfiguration() throws ConfigurationException {
        Parameters params = new Parameters();
        FileBasedConfigurationBuilder<FileBasedConfiguration> confBuilder
                = new FileBasedConfigurationBuilder<FileBasedConfiguration>(
                        PropertiesConfiguration.class)
                .configure(params.properties()
                        .setFileName(PROPERTIES_PATH)
                        .setListDelimiterHandler(new DefaultListDelimiterHandler(','))
                        .setThrowExceptionOnMissing(true));
        FileBasedConfiguration configuration = confBuilder.getConfiguration();
        return configuration;
    }
}
